package indi.zhifa.study2025.flinkcdc;

import com.alibaba.fastjson2.JSONObject;
import indi.zhifa.study2025.commondata.GenItemEntity;
import indi.zhifa.study2025.flinkcdc.binlog.BinlogInfo;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;


public class RedisItemMapper implements RedisMapper<BinlogInfo<GenItemEntity>> {
    @Override
    public RedisCommandDescription getCommandDescription() {
        return new RedisCommandDescription(RedisCommand.SET,500);
    }

    @Override
    public String getKeyFromData(BinlogInfo<GenItemEntity> data) {

        GenItemEntity itemEntity = data.getAfter();
        return "item:"+itemEntity.getId();
    }

    @Override
    public String getValueFromData(BinlogInfo<GenItemEntity> data) {
        GenItemEntity itemEntity = data.getAfter();
        return JSONObject.toJSONString(itemEntity);
    }
}
